package com.xnyzc.lhy.nio.mq;

import com.alibaba.fastjson.JSON;
import com.xnyzc.lhy.common.cache.NettyChannelMap;
import com.xnyzc.lhy.common.component.config.security.RocketConfig;
import com.xnyzc.lhy.common.constant.MethodConstant;
import com.xnyzc.lhy.common.constant.MqConstant;
import com.xnyzc.lhy.common.constant.MqConsumerGroupConstant;
import com.xnyzc.lhy.nio.netty.NettyMsg;
import com.xnyzc.lhy.nio.netty.NettyServerService;
import com.xnyzc.lhy.order.entity.order.OaCOrderPersonal;
import io.netty.channel.ChannelHandlerContext;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.util.List;

@Slf4j
@Component
public class DisOrderMQConsumer {

    /**
     * rocketMq 配置文件
     */
    @Autowired
    RocketConfig rocketConfig;

    /**
     * 消费者
     */
    private DefaultMQPushConsumer consumer;

//    @Value("${rocketMq.isStart}")
//    private boolean isStart;

    /**
     * 开启监听
     */
    @PostConstruct
    public void init() {
        log.info(rocketConfig.getTopic() + "-" + rocketConfig.getNameServerAddr());
//        if (!isStart) {
//            return;
//        }
        try {
            consumer = new DefaultMQPushConsumer(MqConsumerGroupConstant.NOTICE_DIS_ORDER);
            consumer.setNamesrvAddr(rocketConfig.getNameServerAddr());
            consumer.subscribe(rocketConfig.getTopic(), MqConstant.NOTICE_DIS_ORDER);
            //批量消费的最大消息条数
            consumer.setConsumeMessageBatchMaxSize(10);
            //设置消费策略:队列头消费
            consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
            /**
             *  设置监听
             */
            consumer.registerMessageListener(new MessageListenerConcurrently() {
                @Override
                public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                    for (MessageExt msg : msgs) {
                        DisOrderMQConsumer.this.dealBody(msg);
                    }
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }
            });
            consumer.start();
            log.info("---------MQConsumer启动完成------------");
        } catch (Exception e) {
            e.printStackTrace();
            log.info("---------MQConsumer初始化失败------------");
        }
    }

    private void dealBody(MessageExt messageExt) {
        String data = new String(messageExt.getBody());
        log.info("给mq发送的消息为-" + data);
        OaCOrderPersonal personal = JSON.parseObject(data, OaCOrderPersonal.class);
        if (personal == null) {
            return;
        }
        Long driverId = personal.getDriverId();
        String cityId = personal.getCityId();
        ChannelHandlerContext ctx = NettyChannelMap.getData(cityId + ":" + driverId);
        if (ctx != null && !ctx.isRemoved()) {
            NettyServerService.sendMsgToClient(ctx, NettyMsg.setJsonMsg(MethodConstant.DIS_ORDER_NOTICE, data));
        }
    }
}
